package streams.examples.introduction;

import org.apache.kafka.common.serialization.Serde;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.kstream.*;

import java.util.Arrays;
import java.util.Properties;

public class WordCountApplication {


    /**
     * 主程序入口
     * 设置Kafka Streams应用程序的配置，并定义数据处理的拓扑结构
     *
     * @param args 命令行参数
     * @throws Exception 如果应用程序无法正常启动或遇到异常情况
     */
    public static void main(final String[] args) throws Exception {
        // 创建并配置Properties对象，用于设置StreamsConfig的各项配置
        Properties props = new Properties();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "wordcount-application"); // 应用程序ID，用于标识流应用程序
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.0.228:9092"); // Kafka代理地址，应用程序将连接到该地址
        props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass()); // 默认的键序列化和反序列化类
        props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass()); // 默认的值序列化和反序列化类


        // 用于 String 和 Long 类型的序列化器/反序列化器（Serde）
        final Serde<String> stringSerde = Serdes.String();
        final Serde<Long> longSerde = Serdes.Long();

        // 创建StreamsBuilder对象，用于构建流处理拓扑结构
        StreamsBuilder builder = new StreamsBuilder();
        //从输入主题（topic） "streams-plaintext-input" 构建一个 KStream。
        KStream<String, String> textLines = builder.stream(
                "streams-plaintext-input",
                Consumed.with(stringSerde, stringSerde));

        // 处理文本数据，计算每个单词出现的次数，并将结果存储在名为"counts-store"的本地状态存储中
        KTable<String, Long> wordCounts = textLines
                // 将每个文本行按空白字符拆分为单词。
                .flatMapValues(textLine -> Arrays.asList(textLine.toLowerCase().split("\\W+")))
                // 将文本中的单词作为消息键进行分组
                .groupBy((key, word) -> word)
                // 统计每个单词（消息键）的出现次数。
                .count();
//                .count(Materialized.<String, Long, KeyValueStore<Bytes, byte[]>>as("counts-store"));
        // 将实时统计的词频作为变更日志流存储到输出主题中。（topic）"streams-wordcount-output"
        wordCounts.toStream().to("streams-wordcount-output", Produced.with(Serdes.String(), Serdes.Long()));

        // 创建KafkaStreams对象，用于启动和管理流处理应用程序的生命周期
        KafkaStreams streams = new KafkaStreams(builder.build(), props);
        // 启动Kafka Streams应用程序
        streams.start();
    }


}
